import os
import random
import time
from utils import *
from KafkaMsgConsumer import KConsumer
from KafkaMsgProducer import KProducer
import json
import csv
from datetime import datetime
import requests
import threading
from flask import Flask, request, jsonify
from multiprocessing import Process

app = Flask(__name__)

url0 = '10.100.12.46:5001'
eventTypeList = [1, 2]  # 1：逆行，2：障碍物


class TessProcess:
    def __init__(self, param):
        self.Param = param
        self.TESS_RUN = 0
        self.eventTypeList = [1, 2]
        self.perceptionPath = r'E:\test\all_data'
        self.perception_data_path = os.path.join(self.perceptionPath, 'data')
        self.start_flag = 0
        self.back_traffic_flow_flag = 0
        self.target_traffic_flow_flag = 0
        self.event_flag = 0
        self.back_traffic_flow_clear_flag = False
        self.target_traffic_flow_clear_flag = False
        self.start_batch = 0
        self.end_batch = 0
        self.result_data = {}
        self.overNum = 0
        self.filename = 'none'
        self.savePath = r'E:\test\all_data\all_data\save_csv'
        self.bootstrap_servers = '10.102.1.113:9092'

        self.send_traffic_flow = False
        self.send_target_flow = False

        t1 = threading.Thread(target=self.kafka_tess)
        t1.start()

    def kafka_data(self):
        while True:
            if self.TESS_RUN:
                print("1111")
                t2 = threading.Thread(target=self.kafka_configs)
                t2.start()
                # 延时1秒
                time.sleep(1.0)
                t3 = threading.Thread(target=self.kafka_event_body)
                t3.start()
                t4 = threading.Thread(target=self.kafka_traffic)
                t4.start()
                t5 = threading.Thread(target=self.save_kafka_data)
                t5.start()
                break

    def kafka_tess(self):
        user_id = str(self.Param['userId'])
        task_id = str(self.Param['taskId'])
        topic = "{}_{}_tess_config".format(user_id, task_id)
        # bootstrap_servers = '10.102.1.113:9092'
        consumer = KConsumer(topic, self.bootstrap_servers)
        while True:
            message = consumer.sync_consumer()
            json_message_str = message.value.decode('utf-8')
            json_message = json.loads(json_message_str)
            if json_message:
                self.TESS_RUN = json_message['start_flag']
                self.back_traffic_flow_flag = json_message['back_traffic_flow_flag']
                self.target_traffic_flow_flag = json_message['target_traffic_flow_flag']
                self.event_flag = json_message['event_flag']
                self.back_traffic_flow_clear_flag = json_message['back_traffic_flow_clear_flag']
                self.target_traffic_flow_clear_flag = json_message['target_traffic_flow_clear_flag']
                self.start_batch = json_message['start_batch']
                self.end_batch = json_message['end_batch']
            time.sleep(0.5)

    def kafka_configs(self):
        # 加上背景交通流与事件数据流的标识
        user_id = str(self.Param['userId'])
        task_id = str(self.Param['taskId'])
        topic = "{}_{}_config".format(user_id, task_id)
        # bootstrap_servers = '10.102.1.113:9092'
        producer = KProducer(topic, self.bootstrap_servers)
        while True:
            result_data = {'user_id': self.Param['userId'], 'task_id': self.Param['taskId'],
                           'event_type': self.Param['eventType']}
            if self.Param['eventType'] == 2:  # 障碍物
                # 避让数量
                result_data['avoid_number'] = random.randint(2, 10)
                result_data['avoid_behavior'] = self.Param['avoidBehavior']
            else:
                result_data['avoid_number'] = 0
            # 背景车避让距离
            result_data['avoid_distance'] = random.randint(50, 120)
            # 目标出现帧数
            result_data['frame_tate'] = random.randint(15, 20)
            # 背景交通流状态
            result_data['send_traffic_flow'] = self.send_traffic_flow
            # 目标交通流状态
            result_data['send_target_flow'] = self.send_target_flow
            flag = (self.event_flag == 0 and self.overNum == 0) or (self.event_flag == 1 and self.overNum != 0)
            # tess启动且事件结束，重新发送随机配置参数
            if self.TESS_RUN and flag and not self.send_target_flow:
                file_lon = None
                file_lat = None
                course_angle = None
                if self.Param['eventType'] == 1:
                    event_data_path = os.path.join(self.perceptionPath, 'event', 'Nixing_trajectory')
                    event_list = os.listdir(event_data_path)
                    self.filename = random.choice(event_list)
                else:
                    event_data_path = os.path.join(self.perceptionPath, 'event', 'Objects')
                    event_list = os.listdir(event_data_path)
                    self.filename = random.choice(event_list)
                    event_file_path = os.path.join(event_data_path, self.filename)

                    with open(event_file_path, 'r') as file_read:
                        line = file_read.readline()
                        json_file = json.loads(line)
                        participants = json_file['participants']
                        for participant in participants:
                            file_lon = participant['longitude']
                            file_lat = participant['latitude']
                            course_angle = participant['courseAngle']
                send_data = {
                    'user_id': result_data['user_id'],
                    'task_id': result_data['task_id'],
                    'event_type': result_data['event_type'],
                    'avoid_number': result_data['avoid_number'],
                    'avoid_distance': result_data['avoid_distance'],
                    'frame_tate': result_data['frame_tate'],
                    'send_traffic_flow': self.send_traffic_flow,
                    'send_target_flow': self.send_target_flow,
                    'back_traffic_flow_clear_flag': False,
                    'target_traffic_flow_clear_flag': True,
                    'avoid_lon': file_lon,
                    'avoid_lat': file_lat,
                    'avoid_behavior': result_data['avoid_behavior'],
                    'course_angle': course_angle

                }
                # 发送配置项的具体时机：1、当前事件数不大于用户设置事件数量；2、TESS已启动；3、TESS返回事件完成标识
                if self.overNum <= self.Param['eventNum']:
                    print("事件", self.overNum, "的起始帧:", self.start_batch, ", 结束帧：", self.end_batch)
                    print("准备发送配置项，第", self.overNum + 1, "个事件，配置数据为：", send_data)
                    producer.sync_one_producer(send_data)
                    # 发送背景交通流与事件主体交通流标识
                    self.send_traffic_flow = True
                    self.send_target_flow = True
                    # 保存当前发送的配置项内容
                    self.result_data = result_data
                self.overNum += 1
            if self.overNum > self.Param['eventNum']:
                time.sleep(1.0)
                end_tess(result_data['user_id'], result_data['task_id'])
                print("TESS结束")
                break

            time.sleep(1.0)

    def kafka_traffic(self):
        user_id = str(self.Param['userId'])
        task_id = str(self.Param['taskId'])
        topic = "{}_{}_back_traffic_flow".format(user_id, task_id)
        # bootstrap_servers = '10.102.1.113:9092'
        producer = KProducer(topic, self.bootstrap_servers)
        while True:
            # 背景交通流发送具体时机：1、TESS已启动；2、发送配置项后允许发送背景交通流
            if self.TESS_RUN and self.send_traffic_flow:
                print("准备发送背景交通流")
                start_time = self.Param['timeMap'][0]
                end_time = self.Param['timeMap'][1]
                last_time = start_time
                traffic_data_path = os.path.join(self.perception_data_path, 'qifu')
                for filename in os.listdir(traffic_data_path):
                    file_prefix, file_extension = os.path.splitext(filename)
                    if (start_time - 60) <= int(file_prefix) <= end_time:  # 判断文件名是否在时间范围内
                        traffic_path = os.path.join(traffic_data_path, filename)
                        with open(traffic_path, 'r') as f:
                            one_minute_data = [json.loads(line) for line in f]
                        for e1_data in one_minute_data:
                            e1_time = e1_data['timeStamp']
                            e1_start = datetime.strptime(e1_time, "%Y.%m.%d-%H.%M.%S.%f")
                            current_time = e1_start.timestamp()
                            # 如果时间戳位于时间范围内，则处理数据
                            if start_time <= current_time <= end_time:
                                send_data = json.dumps(e1_data)
                                # 发送e1数据
                                producer.sync_one_producer(send_data)
                                step = current_time - last_time
                                last_time = current_time
                                time.sleep(step)
                                if int(current_time) + 1 > end_time:
                                    self.send_traffic_flow = False
            else:
                time.sleep(0.5)

    def kafka_event_body(self):
        user_id = str(self.Param['userId'])
        task_id = str(self.Param['taskId'])
        topic = "{}_{}_target_traffic_flow".format(user_id, task_id)
        # bootstrap_servers = '10.102.1.113:9092'
        producer = KProducer(topic, self.bootstrap_servers)
        while True:
            # print(self.TESS_RUN, self.send_target_flow, self.event_flag)
            # 发送事件主体具体时机：1、TESS已启动；2、发送配置项后允许发送事件主体交通流；3、TESS事件未完成
            if self.TESS_RUN and self.send_target_flow and self.event_flag == 0:
                if self.Param['eventType'] == 1:
                    # last_time = 0.0
                    event_data_path = os.path.join(self.perceptionPath, 'event', 'Nixing_trajectory')
                    traffic_path = os.path.join(event_data_path, self.filename)
                    with open(traffic_path, 'r') as f:
                        all_event_data = [json.loads(line) for line in f]
                    time.sleep(self.result_data['frame_tate'])
                    print("准备发送逆行事件交通流，文件：", self.filename)
                    for e1_event_data in all_event_data:
                        send_data = json.dumps(e1_event_data)
                        # 发送e1数据
                        producer.sync_one_producer(send_data)
                        # current_time = e1_event_data['timeStamp']
                        # step = current_time - last_time
                        # last_time = current_time
                        # time.sleep(step)
                        time.sleep(0.1)
                        if e1_event_data == all_event_data[-1]:
                            self.send_target_flow = False
                else:
                    event_data_path = os.path.join(self.perceptionPath, 'event', 'Objects')
                    traffic_path = os.path.join(event_data_path, self.filename)
                    with open(traffic_path, 'r') as f:
                        all_event_data = [json.loads(line) for line in f]
                    time.sleep(self.result_data['frame_tate'])
                    print("准备发送障碍物事件交通流，文件：", self.filename)
                    send_data = json.dumps(all_event_data[0])
                    producer.sync_one_producer(send_data)
                    time.sleep(0.1)
                    self.send_target_flow = False
            # elif self.event_flag == 1:
            #     self.send_target_flow = False
            else:
                time.sleep(0.5)

    def save_kafka_data(self):
        user_id = str(self.Param['userId'])
        task_id = str(self.Param['taskId'])
        topic = "{}_{}_tess_trajectory".format(user_id, task_id)
        # bootstrap_servers = '10.102.1.113:9092'
        consumer = KConsumer(topic, self.bootstrap_servers)
        while True:
            if self.overNum <= self.Param['eventNum']:
                result_data = self.result_data
                name = str(user_id) + '_' + str(task_id) + "_qifu_" + str(self.Param['timeMap'][0]) + '_' \
                       + str(self.Param['timeMap'][1]) + '_' + str(self.filename.split(".")[0])

                data_save_path = os.path.join(self.savePath, name)
                if not os.path.exists(data_save_path):
                    os.makedirs(data_save_path)
                message = consumer.sync_consumer()
                json_message_ = message.value.decode('utf-8')
                json_message = json.loads(json_message_)
                if json_message:
                    csv_headers = ['id', 'longitude', 'latitude', 'courseAngle', 'length', 'width', 'height', 'speed',
                                   'type']
                    frame = json_message['batchNum']
                    participants = json_message['participants']
                    path = os.path.join(data_save_path, str(frame) + '.csv')
                    with open(path, 'w', newline='') as csvfile:
                        writer = csv.writer(csvfile)
                        writer.writerow(csv_headers)
                        # if participants is not None:
                        for i in range(len(participants)):
                            row = []
                            obj_data = participants[i]
                            row.append(obj_data['id'])
                            row.append(obj_data['longitude'])
                            row.append(obj_data['latitude'])
                            row.append(obj_data['angle'])
                            row.append(obj_data['length'])
                            row.append(obj_data['width'])
                            row.append(obj_data['height'])
                            row.append(obj_data['speed'])
                            row.append(obj_data['type'])
                            writer.writerow(row)
            else:
                time.sleep(0.5)


def http_post(url, data):
    try:
        # 发起POST请求
        headers = {'Content-Type': 'application/json'}
        response = requests.post(url, json=data, headers=headers)
        # 检查响应状态码
        if response.status_code == 200:
            # 请求成功，返回响应内容
            return response.text
        else:
            # 请求失败，打印错误信息
            print("POST请求失败：", response.status_code)
            return None
    except Exception as e:
        # 发生异常，打印异常信息
        print("发生异常：", e)
        return None


def make_response(data, status_code=200):
    response = jsonify(data)
    response.status_code = status_code
    return response


def start_tess(Param):
    url = "http://" + url0 + '/startLocal2Simulator'
    result_data = {
        "userId": Param['userId'],
        "taskId": Param['taskId'],
        "netPath": r'E:\01文档\00仿真项目\数据集\webgl_simulation\webgl\qifusuidao.tess',
        "proj": '+proj=tmerc +lon_0=113.30785068935498 +lat_0=22.97959059227457 +ellps=WGS84'
    }
    print(result_data)
    # print(url)
    response_status = http_post(url, result_data)
    print(response_status)
    if response_status is None:
        return make_response('请求失败', 500)
    else:
        tess_process = TessProcess(Param)
        tess_process.kafka_data()


def end_tess(user_id, task_id):
    # 结束TESS
    url = "http://" + url0 + '/stopSimulator'
    result_data = {
        "userId": user_id,
        "taskId": task_id
    }
    response_status = http_post(url, result_data)
    print(response_status)


@app.route('/start', methods=['Post'])
def get_start_post():
    Param = {}
    param = request.get_json()
    Param['userId'] = param['userId']
    Param['taskId'] = param['taskId']
    Param['eventType'] = param['eventType']
    Param['timeMap'] = [param['start_index'], param['end_index']]
    Param['eventNum'] = param['eventNum']
    assert param['taskId'], '用户id不能为空'
    assert param['eventType'] in eventTypeList, '事件类型不满足要求'
    assert param['taskId'], '样本数不能为空'
    # if not isinstance(param['taskId'], int):
    #     return make_response('样本数必须为整数', 50)
    if not isinstance(param['eventType'], int) and not param['eventType'] in eventTypeList:
        return make_response('事件类型必须为整数，且为' + str(eventTypeList), 500)
    # 车辆激进类型 (0：停车，1: 正常，5：急减速，6：急加速，11：减速待转，12：加速待转)
    if param['eventType'] == 2:
        assert param['avoidBehavior'], '车辆激进类型不能为空'
        Param['avoidBehavior'] = param['avoidBehavior']
    print(Param)

    n = 1
    for i in range(n):  # 创建进程
        # 启动TESS
        p = Process(target=start_tess, args=(Param,))
        p.start()  # 启动进程
        processes_start.append(p)

    return make_response('成功', 200)


@app.route('/end', methods=['Post'])
def get_end_post():
    param = request.get_json()
    userId = param['userId']
    taskId = param['taskId']

    n = 1
    for i in range(n):  # 创建进程
        # 结束TESS
        p = Process(target=end_tess, args=(userId, taskId,))
        p.start()  # 启动进程
        processes_end.append(p)

    return make_response('成功', 200)


def run_start():
    app.run(host='0.0.0.0', port=5002)


def httpserver():
    run_start()


if __name__ == '__main__':
    processes_start = []
    processes_end = []
    httpserver()
